#include <stx/imp/cmn.hpp>

/*
sendmmsg/recvmmsg is Linux-specific. The value specified in vlen is capped to 
UIO_MAXIOV (1024).
*/

inline namespace stx
{
Future<Ips> queryDns(StringView const sv_domain_name, IpType const ip_type)
{
    auto prom      = Promise<Ips>{};
    auto sig_evt   = ::sigevent{};
    auto thrd_attr = ::pthread_attr_t{};
    auto hints     = ::addrinfo{};
    auto ar        = ::gaicb{};
    auto p_ar      = &ar;

    struct CallbackContext_ final
    {
        ::gaicb*      p_ar   = {};
        Promise<Ips>* p_prom = {};
    };

    auto ctx = CallbackContext_{.p_ar = p_ar, .p_prom = &prom};

    ON_SCOPE_EXIT({
        if (ar.ar_result)
        {
            ::freeaddrinfo(ar.ar_result);
        }
    });

    ::pthread_attr_setdetachstate(&thrd_attr, PTHREAD_CREATE_DETACHED);

    if (IpType::K_A == ip_type)
    {
        hints.ai_family = AF_INET;
    }
    else if (IpType::K_AAAA == ip_type)
    {
        hints.ai_family = AF_INET6;
    }
    else
    {
        hints.ai_family = AF_UNSPEC;
    }

    ar.ar_name    = sv_domain_name.data();
    ar.ar_service = nullptr;
    ar.ar_request = &hints;

    sig_evt.sigev_notify                       = SIGEV_THREAD;
    sig_evt.sigev_value.sival_ptr              = &ctx;
    sig_evt._sigev_un._sigev_thread._attribute = &thrd_attr;
    sig_evt._sigev_un._sigev_thread._function  = [](::sigval value)
    {
        auto p_ctx = static_cast<CallbackContext_*>(value.sival_ptr);

        ASSERT(p_ctx);
        ASSERT(p_ctx->p_ar);
        ASSERT(p_ctx->p_prom);

        auto ret = Ips{};
        for (auto result = p_ctx->p_ar->ar_result; result;
             result      = result->ai_next)
        {
            insert(ret, getIpText(*result->ai_addr));
        }

        p_ctx->p_prom->setValue(ret |= actions::sort | actions::unique);
    };

    auto       ips    = Ips{};
    auto const status = ::getaddrinfo_a(GAI_NOWAIT, &p_ar, 1, &sig_evt);
    if (status)
    {
        LOG_ERROR("::getaddrinfo_a: "sv, sv_domain_name, _I_, status);
    }
    else
    {
        ips = co_await prom.getFuture();
    }

    if (ips.empty())
    {
        if (IpType::K_A == ip_type)
        {
            auto shr = co_await shell(
                scat("dig +short "sv, sv_domain_name, " A 2>/dev/null"sv));
            ips = std::move(shr.cout);
        }
        else if (IpType::K_AAAA == ip_type)
        {
            auto shr = co_await shell(
                scat("dig +short "sv, sv_domain_name, " AAAA 2>/dev/null"sv));
            ips = std::move(shr.cout);
        }
        else
        {
            auto shr = co_await shell(
                scat("dig +short "sv, sv_domain_name, " A 2>/dev/null"sv));
            ips = std::move(shr.cout);
            shr = co_await shell(
                scat("dig +short "sv, sv_domain_name, " AAAA 2>/dev/null"sv));
            splice(ips, std::move(shr.cout));
        }
    }

    if (ips.empty())
    {
        if (IpType::K_A == ip_type)
        {
            auto shr = co_await shell(scat(
                "dig @8.8.8.8 +short "sv, sv_domain_name, " A 2>/dev/null"sv));
            ips      = std::move(shr.cout);
        }
        else if (IpType::K_AAAA == ip_type)
        {
            auto shr = co_await shell(scat("dig @8.8.8.8 +short "sv,
                                           sv_domain_name,
                                           " AAAA 2>/dev/null"sv));
            ips      = std::move(shr.cout);
        }
        else
        {
            auto shr = co_await shell(scat(
                "dig @8.8.8.8 +short "sv, sv_domain_name, " A 2>/dev/null"sv));

            ips = std::move(shr.cout);
            shr = co_await shell(scat("dig @8.8.8.8 +short "sv,
                                      sv_domain_name,
                                      " AAAA 2>/dev/null"sv));
            splice(ips, std::move(shr.cout));
        }
    }

    if (ips.empty())
    {
        LOG_WARN("Failed to query DNS: "sv, sv_domain_name);
    }

    auto ret = Ips{};
    for (auto&& s_ip : ips)
    {
        if (isIp4(s_ip) || isIp6(s_ip))
        {
            insert(ret, std::move(s_ip));
        }
    }

    co_return ret;
}

DnsCache& DnsCache::instance() noexcept
{
    static auto ls = DnsCache{};
    return ls;
}

Future<Ips> DnsCache::query(StringView const    sv_domain_name,
                            IpType const        ip_type,
                            SbUpdateCache const sb_update_cache)
{
    auto key = Key_{.ip_type = ip_type, .domain_name = tos(sv_domain_name)};

    if (!sb_update_cache)
    {
        auto opt_ips = this->atomic_dict_.visit(
            [&](auto const& dict) -> std::optional<Ips>
            {
                auto const pos = dict.find(key);
                if (pos != dict.end())
                {
                    return pos->second;
                }

                return std::nullopt;
            });

        if (opt_ips && !opt_ips->empty())
        {
            co_return std::move(*opt_ips);
        }
    }

    auto ips = co_await queryDns(sv_domain_name, ip_type);
    this->atomic_dict_.visit([&](auto& dict) { dict[std::move(key)] = ips; });
    if (ips.empty())
    {
        LOG_ERROR("Failed to queryDns: "sv, ip_type, _I_, sv_domain_name);
    }

    co_return ips;
}

std::optional<Ip4> getIp4(char const* const sz_ip4)
{
    auto       ret    = Ip4{};
    auto const status = ::inet_pton(AF_INET, sz_ip4, ret.data());
    if (1 != status)
    {
        LOG_ERROR(
            "::inet_pton(AF_INET): "sv, status, _I_, sz_ip4, _I_, M_SYS_ERR);
        return {};
    }

    return ret;
}

std::optional<Ip6> getIp6(char const* const sz_ip6)
{
    auto       ret    = Ip6{};
    auto const status = ::inet_pton(AF_INET6, sz_ip6, ret.data());
    if (1 != status)
    {
        LOG_ERROR(
            "::inet_pton(AF_INET6): "sv, status, _I_, sz_ip6, _I_, M_SYS_ERR);
        return {};
    }

    return ret;
}

Ip4 operator""_ip4(char const* const sz, std::size_t const)
{
    return *getIp4(sz);
}

Ip6 operator""_ip6(char const* const sz, std::size_t const)
{
    return *getIp6(sz);
}

[[nodiscard]] Future<Endpoint> getEndpoint(std::string const s_addr,
                                           Port const        port)
{
    try
    {
        if (isIp4(s_addr))
        {
            auto const opt_ip4 = getIp4(s_addr.data());
            if (!opt_ip4)
            {
                LOG_ERROR("Failed to getIp4: "sv, s_addr);
                co_return Endpoint{};
            }

            co_return Endpoint(*opt_ip4, port);
        }

        if (isIp6(s_addr))
        {
            auto const opt_ip6 = getIp6(s_addr.data());
            if (!opt_ip6)
            {
                LOG_ERROR("Failed to getIp6: "sv, s_addr);
                co_return Endpoint{};
            }

            co_return Endpoint(*opt_ip6, port);
        }

        auto const ips = co_await DnsCache::instance().query(s_addr);
        if (ips.empty())
        {
            LOG_ERROR("Invalid sv_ep: "sv, s_addr);
            co_return Endpoint{};
        }

        auto const& ip_text = ips[rand(0uz, ips.size() - 1)];
        if (isIp4(ip_text))
        {
            auto const opt_ip4 = getIp4(ip_text.data());
            if (!opt_ip4)
            {
                LOG_ERROR("Failed to getIp4: "sv, s_addr, _I_, ip_text);
                co_return Endpoint{};
            }

            co_return Endpoint(*opt_ip4, port);
        }

        if (isIp6(ip_text))
        {
            auto const opt_ip6 = getIp6(ip_text.data());
            if (!opt_ip6)
            {
                LOG_ERROR("Failed to getIp6: "sv, s_addr, _I_, ip_text);
                co_return Endpoint{};
            }

            co_return Endpoint(*opt_ip6, port);
        }
    }
    catch (std::exception const& e)
    {
        LOG_ERROR(e.what(), _I_, s_addr, _I_, port);
    }

    LOG_ERROR("Failed to getEndpoint from: "sv, s_addr, _I_, port);
    co_return Endpoint{};
}

[[nodiscard]] Future<Endpoint> getEndpoint(std::string_view const sv_ep)
{
    try
    {
        auto const mr =
            getMatchResults(sv_ep, rgx<R"(^(.+):([1-9][0-9]*)$)">());
        if (mr.size() != 3)
        {
            LOG_ERROR("Invalid sv_ep: "sv, sv_ep);
            co_return Endpoint{};
        }

        if (mr[1].size() < 2uz)
        {
            LOG_ERROR("Invalid sv_ep: "sv, sv_ep);
            co_return Endpoint{};
        }

        auto       s_addr   = tos(mr[1]);
        auto const opt_port = ton(mr[2]);
        if (!opt_port)
        {
            LOG_ERROR("Invalid sv_ep: "sv, sv_ep);
            co_return Endpoint{};
        }

        if (s_addr.size() > 2 && '[' == s_addr.front() && ']' == s_addr.back())
        {
            s_addr = std::string(s_addr.data() + 1, s_addr.size() - 2);
        }

        co_return co_await getEndpoint(std::move(s_addr), Port(*opt_port));
    }
    catch (std::exception const& e)
    {
        LOG_ERROR(e.what());
    }

    LOG_ERROR("Failed to getEndpoint from: "sv, sv_ep);
    co_return Endpoint{};
}

Endpoint::Endpoint(::sockaddr const& skt_addr)
{
    if (AF_INET == skt_addr.sa_family)
    {
        *this = Endpoint(*std::bit_cast<::sockaddr_in const*>(&skt_addr));
    }
    else if (AF_INET6 == skt_addr.sa_family)
    {
        *this = Endpoint(*std::bit_cast<::sockaddr_in6 const*>(&skt_addr));
    }
    else
    {
        RUNTIME_ERROR("Invalid sa_family: "sv, skt_addr.sa_family);
    }
}

Endpoint::Endpoint(::sockaddr_in const& skt_addr4) noexcept
    : skt_addr_{.skt_addr4 = skt_addr4}
{}

Endpoint::Endpoint(::sockaddr_in6 const& skt_addr6) noexcept
    : skt_addr_{.skt_addr6 = skt_addr6}
{}

Endpoint::Endpoint(Ip4 const ip4, Port const port) noexcept : Endpoint{}
{
    this->skt_addr_.skt_addr4.sin_family = AF_INET;
    this->skt_addr_.skt_addr4.sin_port   = ::htons(port.value);
    std::memcpy(&this->skt_addr_.skt_addr4.sin_addr, ip4.data(), 4);
}

Endpoint::Endpoint(Ip6 const& ip6, Port const port) noexcept
{
    this->skt_addr_.skt_addr6.sin6_family = AF_INET6;
    this->skt_addr_.skt_addr6.sin6_port   = ::htons(port.value);
    std::memcpy(&this->skt_addr_.skt_addr6.sin6_addr, ip6.data(), 16);
}

Ip4 Endpoint::ip4() const
{
    if (AF_INET == this->skt_addr_.skt_addr.sa_family)
    {
        auto ret = Ip4{};
        std::memcpy(ret.data(), &this->skt_addr_.skt_addr4.sin_addr, 4);
        return ret;
    }

    RUNTIME_ERROR("Endpoint is not ip4."sv);
}

Ip6 Endpoint::ip6() const
{
    if (AF_INET6 == this->skt_addr_.skt_addr.sa_family)
    {
        auto ret = Ip6{};
        std::memcpy(ret.data(), &this->skt_addr_.skt_addr6.sin6_addr, 16);
        return ret;
    }

    RUNTIME_ERROR("Endpoint is not ip6."sv);
}

std::string tos(Port const port) noexcept
{
    return tos(port.value);
}

std::string tos(Ip4 const ip4)
{
    auto buf = std::array<char, INET6_ADDRSTRLEN>{};
    if (::inet_ntop(AF_INET, ip4.data(), buf.data(), buf.size()))
    {
        return buf.data();
    }

    LOG_ERROR("::inet_ntop(AF_INET): "sv, bin2hex(ip4));
    return {};
}

std::string tos(Ip6 const& ip6)
{
    auto buf = std::array<char, INET6_ADDRSTRLEN>{};
    if (::inet_ntop(AF_INET6, ip6.data(), buf.data(), buf.size()))
    {
        return buf.data();
    }

    LOG_ERROR("::inet_ntop(AF_INET6): "sv, bin2hex(ip6));
    return {};
}

std::string tos(Endpoint const& ep)
{
    if (ep.isIp4())
    {
        return scat(ep.ip4(), ":"sv, ep.port());
    }
    else if (ep.isIp6())
    {
        return scat("["sv, ep.ip6(), "]:"sv, ep.port());
    }

    LOG_ERROR("Invalid ep."sv);
    return {};
}

std::string getIpText(::sockaddr const& skt_addr)
{
    auto buf = std::array<char, INET6_ADDRSTRLEN>{};
    if (AF_INET == skt_addr.sa_family)
    {
        auto const sin = std::bit_cast<::sockaddr_in*>(&skt_addr);
        if (!::inet_ntop(AF_INET, &sin->sin_addr, buf.data(), buf.size()))
        {
            LOG_ERROR("::inet_ntop(AF_INET): "sv, bin2hex(skt_addr));
            return {};
        }
    }
    else if (AF_INET6 == skt_addr.sa_family)
    {
        auto const sin6 = std::bit_cast<::sockaddr_in6*>(&skt_addr);
        if (!::inet_ntop(AF_INET6, &sin6->sin6_addr, buf.data(), buf.size()))
        {
            LOG_ERROR("::inet_ntop(AF_INET6): "sv, bin2hex(skt_addr));
            return {};
        }
    }
    else
    {
        LOG_ERROR("Unknown sa_family: "sv, skt_addr.sa_family);
        return {};
    }

    return buf.data();
}

std::string getPortText(::sockaddr const& skt_addr)
{
    if (AF_INET == skt_addr.sa_family)
    {
        auto const sin = std::bit_cast<::sockaddr_in*>(&skt_addr);
        return tos(std::byteswap(sin->sin_port));
    }
    else if (AF_INET6 == skt_addr.sa_family)
    {
        auto const sin6 = std::bit_cast<::sockaddr_in6*>(&skt_addr);
        return tos(std::byteswap(sin6->sin6_port));
    }

    LOG_ERROR("Unknown sa_family: "sv, skt_addr.sa_family);
    return {};
}

bool Socket::resizeRecvBuf(std::size_t const new_size) noexcept
{
    auto opt = static_cast<int>(std::min(new_size, 1_Gi));
    if (::setsockopt(
            this->fd_, SOL_SOCKET, SO_RCVBUF, &opt, Number(sizeof(opt))))
    {
        LOG_ERROR("::setsockopt(SO_RCVBUF): "sv, new_size, _I_, M_SYS_ERR);
        return false;
    }

    return true;
}

bool Socket::resizeSendBuf(std::size_t const new_size) noexcept
{
    auto opt = static_cast<int>(std::min(new_size, 1_Gi));
    if (::setsockopt(
            this->fd_, SOL_SOCKET, SO_SNDBUF, &opt, Number(sizeof(opt))))
    {
        LOG_ERROR("::setsockopt(SO_SNDBUF): "sv, new_size, _I_, M_SYS_ERR);
        return false;
    }

    return true;
}

bool Socket::bind(Endpoint const& local_ep) noexcept
{
    if (this->fd_ < 0)
    {
        LOG_ERROR(this->fd_, _I_, local_ep);
        return false;
    }

    auto const status =
        ::bind(this->fd_, local_ep.getSockAddr(), local_ep.getSockLen());
    if (status < 0)
    {
        LOG_WARN("::bind: "sv, this->fd_, _I_, local_ep, _I_, M_SYS_ERR);
        return false;
    }

    return true;
}

Future<Status> Socket::connect(Endpoint const& remote_ep) noexcept
{
    if (this->fd_ < 0)
    {
        LOG_ERROR(this->fd_, _I_, remote_ep);
        co_return C_INVALID_FD;
    }

    auto const status =
        ::connect(this->fd_, remote_ep.getSockAddr(), remote_ep.getSockLen());
    if (status < 0)
    {
        if (EINPROGRESS == errno)
        {
            auto const fd_state =
                co_await wait(this->fd_, FdState::K_WRITABLE, 30s);
            if (!isWritable(fd_state))
            {
                LOG_ERROR("Invalid fd_state: "sv, fd_state, _I_, remote_ep);
                co_return C_INVALID_FD_STATE;
            }

            auto sys_errno = int{};
            auto len       = ::socklen_t{};
            len            = Number(sizeof(sys_errno));

            if (::getsockopt(this->fd_, SOL_SOCKET, SO_ERROR, &sys_errno, &len))
            {
                auto const sys_errno1 = -errno;
                LOG_ERROR(
                    "::getsockopt(SO_ERROR): "sv, remote_ep, _I_, M_SYS_ERR);
                co_return sys_errno1;
            }

            ASSERT(sys_errno >= 0);
            if (sys_errno)
            {
                LOG_ERROR("System error: "sv,
                          remote_ep,
                          ""sv,
                          sys_errno,
                          _I_,
                          std::strerror(sys_errno));
                co_return -sys_errno;
            }

            co_return 0;
        }

        auto const sys_errno = -errno;
        LOG_ERROR("::connect: "sv, this->fd_, _I_, remote_ep, _I_, M_SYS_ERR);
        co_return sys_errno;
    }

    co_return status;
}

bool Socket::initServer_(Endpoint const&   local_ep,
                         SbReusePort const sb_reuse_port) noexcept
{
    // Fo a server socket, SO_REUSEADDR is good and safe, so always enable it.
    auto opt = 1;
    if (::setsockopt(
            this->fd_, SOL_SOCKET, SO_REUSEADDR, &opt, Number(sizeof(opt))))
    {
        auto const sys_errno = -errno;
        LOG_ERROR("::setsockopt(SO_REUSEADDR): "sv,
                  this->fd_,
                  _I_,
                  local_ep,
                  _I_,
                  M_SYS_ERR);
        this->close();
        this->fd_ = sys_errno;
        return false;
    }

    if (sb_reuse_port)
    {
        opt = 1;
        if (::setsockopt(
                this->fd_, SOL_SOCKET, SO_REUSEPORT, &opt, Number(sizeof(opt))))
        {
            auto const sys_errno = -errno;
            LOG_ERROR("::setsockopt(SO_REUSEPORT): "sv,
                      this->fd_,
                      _I_,
                      local_ep,
                      _I_,
                      M_SYS_ERR);
            this->close();
            this->fd_ = sys_errno;
            return false;
        }
    }

    if (!this->bind(local_ep))
    {
        auto const sys_errno = -errno;
        LOG_WARN("this->bind: "sv, local_ep, _I_, M_SYS_ERR);
        this->close();
        this->fd_ = sys_errno;
        return false;
    }

    return true;
}

Future<RecvMsgResult> UdpSocket::recv(MemSpan const  ms,
                                      Duration const dur_timeout)
{
    if (this->fd_ < 0)
    {
        LOG_ERROR(this->fd_);
        co_return RecvMsgResult{.status = C_INVALID_FD};
    }

    if (ms.empty())
    {
        LOG_ERROR("No buffer to recv msg."sv);
        co_return RecvMsgResult{.status = C_INVALID_OUTPUT_SPAN};
    }

    auto skt_addr = ::sockaddr_in6{};
    auto hdr      = ::msghdr{};
    auto iov      = std::array<::iovec, 1>{};

    iov[0].iov_base = ms.data();
    iov[0].iov_len  = ms.size();

    hdr.msg_iov     = iov.data();
    hdr.msg_iovlen  = 1;
    hdr.msg_name    = &skt_addr;
    hdr.msg_namelen = Number(sizeof(skt_addr));

    auto const tp_start = SteadyClock::now();

    do
    {
        auto const status = ::recvmsg(this->fd_, &hdr, MSG_DONTWAIT);
        if (status >= 0)
        {
            co_return RecvMsgResult{
                .msg = {.ms        = MemSpan(iov[0].iov_base, status),
                        .remote_ep = Endpoint(
                            *std::bit_cast<::sockaddr const*>(hdr.msg_name))}};
        }

        if (EAGAIN == errno)
        {
            if (dur_timeout > 0ms && elapsed(tp_start) >= dur_timeout)
            {
                co_return RecvMsgResult{.status = C_IO_TIMEOUT};
            }

            auto const fd_state =
                co_await wait(this->fd_, FdState::K_READABLE, dur_timeout);
            if (isReadable(fd_state))
            {
                continue;
            }

            LOG_ERROR("Invalid fd_state: "sv, fd_state);
            co_return RecvMsgResult{.status = C_INVALID_FD_STATE};
        }

        auto const sys_errno = -errno;
        LOG_ERROR(this->fd_, _I_, M_SYS_ERR);
        co_return RecvMsgResult{.status = sys_errno};
    } while (true);
}

Future<RecvMsgsResult> UdpSocket::recv(std::vector<MemSpan> const mss,
                                       Duration const             dur_timeout)
{
    if (this->fd_ < 0)
    {
        LOG_ERROR(this->fd_);
        co_return RecvMsgsResult{.status = C_INVALID_FD};
    }

    if (std::empty(mss))
    {
        LOG_ERROR("No buffer to recv msgs."sv);
        co_return RecvMsgsResult{.status = C_INVALID_OUTPUT_SPAN};
    }

    auto skt_addrs = PodBuffer<::sockaddr_in6>(std::size(mss));
    auto hdrs      = PodBuffer<::mmsghdr>(std::size(mss));
    auto iovec     = PodBuffer<::iovec>(std::size(mss));

    zero(skt_addrs);
    zero(hdrs);
    zero(iovec);

    FOR (i, auto const ms : mss)
    {
        auto& hdr               = hdrs[i];
        iovec[i].iov_base       = ms.data();
        iovec[i].iov_len        = ms.size();
        hdr.msg_hdr.msg_iov     = &iovec[i];
        hdr.msg_hdr.msg_iovlen  = 1;
        hdr.msg_hdr.msg_name    = &skt_addrs[i];
        hdr.msg_hdr.msg_namelen = Number(sizeof(::sockaddr_in6));
    }

    auto const tp_start = SteadyClock::now();

    do
    {
        auto const status = ::recvmmsg(this->fd_,
                                       hdrs.data(),
                                       Number(hdrs.size()),
                                       MSG_DONTWAIT | MSG_WAITFORONE,
                                       nullptr);
        if (status >= 0)
        {
            auto ret = RecvMsgsResult{};
            ret.msgs.reserve(Number(status));

            LOOP (i, static_cast<std::size_t>(status))
            {
                ret.total_size += hdrs[i].msg_len;

                auto ms        = MemSpan(iovec[i].iov_base, hdrs[i].msg_len);
                auto remote_ep = Endpoint(skt_addrs[i]);
                auto incomming_msg = IncomingMsg{
                    .ms = std::move(ms), .remote_ep = std::move(remote_ep)};

                insert(ret.msgs, std::move(incomming_msg));
            }

            co_return ret;
        }

        if (EAGAIN == errno)
        {
            if (dur_timeout > 0ms && elapsed(tp_start) >= dur_timeout)
            {
                co_return RecvMsgsResult{.status = C_IO_TIMEOUT};
            }

            auto const fd_state =
                co_await wait(this->fd_, FdState::K_READABLE, dur_timeout);
            if (isReadable(fd_state))
            {
                continue;
            }

            LOG_ERROR("Invalid fd_state: "sv, fd_state);
            co_return RecvMsgsResult{.status = C_INVALID_FD_STATE};
        }

        auto const sys_errno = -errno;
        LOG_ERROR(this->fd_, _I_, M_SYS_ERR);
        co_return RecvMsgsResult{.status = sys_errno};
    } while (true);
}

Future<std::ssize_t> UdpSocket::send(MemView const  mv,
                                     Duration const dur_timeout)
{
    if (this->fd_ < 0)
    {
        LOG_ERROR(this->fd_);
        co_return C_INVALID_FD;
    }

    if (mv.empty())
    {
        LOG_ERROR("No msg to send."sv);
        co_return C_INVALID_INPUT_VIEW;
    }

    auto hdr = ::msghdr{};
    auto iov = std::array<::iovec, 1>{};

    iov[0].iov_base = std::bit_cast<void*>(mv.data());
    iov[0].iov_len  = mv.size();

    hdr.msg_iov    = iov.data();
    hdr.msg_iovlen = 1;

    auto const tp_start = SteadyClock::now();

    do
    {
        auto const status = ::sendmsg(this->fd_, &hdr, MSG_DONTWAIT);
        if (status >= 0)
        {
            co_return status;
        }

        if (EAGAIN == errno)
        {
            if (dur_timeout > 0ms && elapsed(tp_start) >= dur_timeout)
            {
                co_return C_IO_TIMEOUT;
            }

            auto const fd_state =
                co_await wait(this->fd_, FdState::K_WRITABLE, dur_timeout);
            if (isWritable(fd_state))
            {
                continue;
            }

            LOG_ERROR("Invalid fd_state: "sv, fd_state);
            co_return C_INVALID_FD_STATE;
        }

        auto const sys_errno = -errno;
        LOG_ERROR(this->fd_, _I_, M_SYS_ERR);
        co_return sys_errno;
    } while (true);
}

Future<std::ssize_t>
UdpSocket::send(IsOutgoingMsgRange auto const outgoing_msgs,
                Duration const                dur_timeout)
{
    if (this->fd_ < 0)
    {
        LOG_ERROR(this->fd_);
        co_return C_INVALID_FD;
    }

    if (outgoing_msgs.empty())
    {
        LOG_ERROR("No msgs to send."sv);
        co_return C_INVALID_INPUT_VIEW;
    }

    auto hdrs  = PodBuffer<::mmsghdr>(outgoing_msgs.size());
    auto iovec = PodBuffer<::iovec>(outgoing_msgs.size());

    zero(hdrs);
    zero(iovec);

    LOOP (i, outgoing_msgs.size())
    {
        auto&       hdr          = hdrs[i];
        auto const& outgoing_msg = outgoing_msgs[i];

        iovec[i].iov_base       = std::bit_cast<void*>(outgoing_msg.mv.data());
        iovec[i].iov_len        = outgoing_msg.mv.size();
        hdr.msg_hdr.msg_iov     = &iovec[i];
        hdr.msg_hdr.msg_iovlen  = 1;
        hdr.msg_hdr.msg_name    = outgoing_msg.remote_ep.getSockAddr();
        hdr.msg_hdr.msg_namelen = outgoing_msg.remote_ep.getSockLen();
    }

    auto const tp_start = SteadyClock::now();

    do
    {
        auto const status = ::sendmmsg(
            this->fd_, hdrs.data(), Number(hdrs.size()), MSG_DONTWAIT);
        if (status >= 0)
        {
            co_return status;
        }

        if (EAGAIN == errno)
        {
            if (dur_timeout > 0ms && elapsed(tp_start) >= dur_timeout)
            {
                co_return C_IO_TIMEOUT;
            }

            auto const fd_state =
                co_await wait(this->fd_, FdState::K_WRITABLE, dur_timeout);
            if (isWritable(fd_state))
            {
                continue;
            }

            LOG_ERROR("Invalid fd_state: "sv, fd_state);
            co_return C_INVALID_FD_STATE;
        }

        auto const sys_errno = -errno;
        LOG_ERROR(this->fd_, _I_, M_SYS_ERR);
        co_return sys_errno;
    } while (true);
}

Future<std::ssize_t> UdpSocket::send(OutgoingMsg    outgoing_msg,
                                     Duration const dur_timeout)
{
    auto outgoing_msgs = std::array<OutgoingMsg, 1>{};
    outgoing_msgs[0]   = std::move(outgoing_msg);
    co_return co_await this->send(outgoing_msgs, dur_timeout);
}

bool TcpSocket::disableTimeWait() noexcept
{
    auto opt     = ::linger{};
    opt.l_onoff  = true;
    opt.l_linger = 0;

    if (::setsockopt(
            this->fd_, SOL_SOCKET, SO_LINGER, &opt, Number(sizeof(opt))))
    {
        LOG_ERROR("::setsockopt(SO_LINGER): "sv, this->fd_, _I_, M_SYS_ERR);
        return false;
    }

    return true;
}

bool TcpSocket::enableNoDealy() noexcept
{
    auto opt = 1;
    if (::setsockopt(
            this->fd_, SOL_SOCKET, TCP_NODELAY, &opt, Number(sizeof(opt))))
    {
        LOG_ERROR("::setsockopt(TCP_NODELAY): "sv, this->fd_, _I_, M_SYS_ERR);
        return false;
    }

    return true;
}

bool TcpSocket::enableCork() noexcept
{
    auto opt = 1;
    if (::setsockopt(
            this->fd_, SOL_SOCKET, TCP_CORK, &opt, Number(sizeof(opt))))
    {
        LOG_ERROR("::setsockopt(TCP_CORK): "sv, this->fd_, _I_, M_SYS_ERR);
        return false;
    }

    return true;
}

TcpServer::TcpServer(Endpoint             local_ep,
                     TcpBacklogSize const backlog_size,
                     SbReusePort const    sb_reuse_port)
{
    auto domain = int{};

    if (local_ep.isIp4())
    {
        domain = AF_INET;
    }
    else if (local_ep.isIp6())
    {
        domain = AF_INET6;
    }
    else
    {
        LOG_ERROR("Invalid local_ep: "sv, local_ep);
        return;
    }

    this->fd_ = ::socket(domain, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    if (this->fd_ < 0)
    {
        this->fd_ = -errno;
        LOG_ERROR("::socket(SOCK_STREAM): "sv, local_ep, _I_, M_SYS_ERR);
        return;
    }

    if (!this->initServer_(local_ep, sb_reuse_port))
    {
        LOG_WARN("this->initServer_: "sv, local_ep, _I_, M_SYS_ERR);
        return;
    }

    if (::listen(this->fd_, static_cast<int>(backlog_size)))
    {
        auto const sys_errno = -errno;
        LOG_ERROR(
            "this->listen: "sv, local_ep, _I_, backlog_size, _I_, M_SYS_ERR);
        this->close();
        this->fd_ = sys_errno;
        return;
    }
}

Future<TcpSocket> TcpServer::accept(Endpoint* const p_remote_ep) noexcept
{
    auto buf     = ::sockaddr_in6{};
    auto buf_len = ::socklen_t{};

    buf_len = Number(sizeof(buf));

    auto p_skt_addr = Ptr<::sockaddr>{};
    auto p_skt_len  = Ptr<::socklen_t>{};

    if (p_remote_ep)
    {
        p_skt_addr = std::bit_cast<::sockaddr*>(&buf);
        p_skt_len  = &buf_len;
    }

    do
    {
        auto const fd = ::accept4(
            this->fd_, p_skt_addr, p_skt_len, SOCK_NONBLOCK | SOCK_CLOEXEC);
        if (fd < 0)
        {
            if (EAGAIN == errno)
            {
                auto const fd_state =
                    co_await wait(this->fd_, FdState::K_READABLE);
                if (isReadable(fd_state))
                {
                    continue;
                }

                LOG_ERROR("Invalid fd_state: "sv, fd_state);
                co_return TcpSocket(C_INVALID_FD_STATE);
            }

            auto const sys_errno = -errno;
            LOG_ERROR("::accept4: "sv, this->fd_, _I_, M_SYS_ERR);
            co_return TcpSocket(sys_errno);
        }

        if (p_remote_ep)
        {
            *p_remote_ep = Endpoint(*p_skt_addr);
        }

        co_return TcpSocket(fd);
    } while (true);
}

UdpServer::UdpServer(Endpoint const& local_ep, SbReusePort const sb_reuse_port)
{
    auto domain = int{};

    if (local_ep.isIp4())
    {
        domain = AF_INET;
    }
    else if (local_ep.isIp6())
    {
        domain = AF_INET6;
    }
    else
    {
        LOG_ERROR("Invalid local_ep: "sv, local_ep);
        this->fd_ = C_INVALID_LOCAL_EP;
        return;
    }

    this->fd_ = ::socket(domain, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    if (this->fd_ < 0)
    {
        this->fd_ = -errno;
        LOG_ERROR("::socket(SOCK_DGRAM): "sv, local_ep, _I_, M_SYS_ERR);
        return;
    }

    if (!this->initServer_(local_ep, sb_reuse_port))
    {
        LOG_ERROR("this->initServer_: "sv, local_ep, _I_, M_SYS_ERR);
    }
}

Future<> startTcpServer(Endpoint const       local_ep,
                        FnOnTcpNewConn       fn_on_new_conn,
                        TcpBacklogSize const backlog_size,
                        SbReusePort const    sb_reuse_port)
{
    auto tcp_svr = TcpServer(local_ep, backlog_size, sb_reuse_port);

    while (tcp_svr)
    {
        auto remote_ep   = Endpoint{};
        auto tcp_svr_skt = co_await tcp_svr.accept(&remote_ep);
        if (tcp_svr_skt)
        {
            fn_on_new_conn(
                std::move(tcp_svr_skt), std::move(remote_ep), local_ep);
            continue;
        }

        LOG_ERROR("Failed to accept tcp connection: "sv,
                  tcp_svr_skt.fd(),
                  _I_,
                  local_ep);
    }
}

Future<> startTcpServer(Endpoint local_ep, FnOnTcpNewConn fn_on_new_conn)
{
    co_return co_await startTcpServer(
        std::move(local_ep), std::move(fn_on_new_conn), {}, {});
}

Future<TcpSocket> startTcpClient(Endpoint const remote_ep)
{
    auto domain = int{};
    if (remote_ep.isIp4())
    {
        domain = AF_INET;
    }
    else if (remote_ep.isIp6())
    {
        domain = AF_INET6;
    }
    else
    {
        LOG_ERROR("Invalid remote_ep: "sv, remote_ep);
        co_return TcpSocket(C_INVALID_REMOTE_EP);
    }

    auto const fd =
        ::socket(domain, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    if (fd < 0)
    {
        auto const sys_errno = -errno;
        LOG_ERROR("::socket(SOCK_STREAM): "sv, remote_ep, _I_, M_SYS_ERR);
        co_return TcpSocket(sys_errno);
    }

    auto       tcp_skt = TcpSocket(fd);
    auto const status  = co_await tcp_skt.connect(remote_ep);
    if (C_STATUS_SUCCESS == status)
    {
        co_return tcp_skt;
    }

    LOG_ERROR("tcp_skt.connect: "sv, fd, _I_, status, _I_, remote_ep);
    co_return TcpSocket(status);
}

Future<UdpSocket> startUdpClient(Endpoint const remote_ep)
{
    auto domain = int{};

    if (remote_ep.isIp4())
    {
        domain = AF_INET;
    }
    else if (remote_ep.isIp6())
    {
        domain = AF_INET6;
    }
    else
    {
        LOG_ERROR("Invalid remote_ep: "sv, remote_ep);
        co_return UdpSocket(C_INVALID_REMOTE_EP);
    }

    auto const fd =
        ::socket(domain, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    if (fd < 0)
    {
        auto const sys_errno = -errno;
        LOG_ERROR("::socket(SOCK_DGRAM): "sv, remote_ep, _I_, M_SYS_ERR);
        co_return UdpSocket(sys_errno);
    }

    auto       udp_skt = UdpSocket(fd);
    auto const status  = co_await udp_skt.connect(remote_ep);
    if (C_STATUS_SUCCESS == status)
    {
        co_return udp_skt;
    }

    LOG_ERROR("udp_skt.connect: "sv, fd, _I_, status, _I_, remote_ep);
    co_return UdpSocket(status);
}

Future<TcpSocket> startTcpClient(std::string_view const sv_addr,
                                 Port const             port)
{
    auto remote_ep = co_await getEndpoint(tos(sv_addr), port);
    if (!remote_ep)
    {
        LOG_ERROR("Failed to getEndpoint: "sv, sv_addr, _I_, port);
        co_return TcpSocket{};
    }

    co_return co_await startTcpClient(std::move(remote_ep));
}

Future<TcpSocket> startTcpClient(std::string_view const sv_ep)
{
    auto remote_ep = co_await getEndpoint(sv_ep);
    if (!remote_ep)
    {
        LOG_ERROR("Failed to getEndpoint: "sv, sv_ep);
        co_return TcpSocket{};
    }

    co_return co_await startTcpClient(std::move(remote_ep));
}
} // namespace stx

/*
The msg_name field of struct msghdr does not necessarily have to point to a
struct sockaddr_in: it points to a generic socket address; the exact structure 
depends on the socket family: if it's an AF_UNIX socket, it points to a struct 
sockaddr_un, if it's AF_INET, it points to struct sockaddr_in, and if it's 
AF_INET6 it points to struct sockaddr_in6. All of them share the sa_family 
field, which is also the first field, so you can cast msg_name to struct 
sockaddr *, read the sa_family field, and decide where to go from there.

Keep in mind that there may be other socket families implemented; Linux, for 
example, has AF_APPLETALK, AF_X25, and some others I don't remember right now.

Also, the msg->msg_name in my sendmsg function always equals to sizeof(struct 
sockaddr_in).

This is true as long as you're using AF_INET sockets, because ip addresses and 
port numbers have a fixed length. But the size argument is not always a 
statically known value. In UNIX sockets (local sockets), the size you pass is 
not sizeof(struct sockaddr_un), but instead it is offsetof(struct sockaddr_un,
sa_data)+strlen(pathname), where pathname is the size of the path string in the
filesystem where the socket handle lives.

sendmsg can use IP_PKTINFO ancillary data to specify the interface to send data.
*/